package com.googlecode.quillen.service;

import com.googlecode.quillen.domain.*;
import static com.googlecode.quillen.util.Utils.logDebug;
import com.googlecode.quillen.util.Pair;
import com.googlecode.quillen.util.ResultConsumer;
import com.googlecode.quillen.repository.ObjectStorage;
import com.googlecode.quillen.repository.AttributeStorage;
import com.googlecode.quillen.repository.AttributeStorageQuery;

import java.io.IOException;
import java.io.InputStream;
import java.util.*;
import java.security.NoSuchAlgorithmException;
import java.security.MessageDigest;
import java.text.SimpleDateFormat;
import java.text.ParseException;

import org.apache.commons.codec.binary.Hex;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;

/**
 * Created by IntelliJ IDEA.
 * User: greg
 * Date: Dec 6, 2008
 * Time: 3:28:07 PM
 */
public class ShadowFileServiceImpl implements ShadowFileService
{
    private static final Log LOG = LogFactory.getLog(ShadowFileServiceImpl.class);
    private static final String SNAPSHOT_DATE_DELIM = "#";

    private final AttributeStorage attributeStorage;
    private final ObjectStorage objectStorage;
    private final CompressionService compressionService;

    public ShadowFileServiceImpl(CompressionService compressionService, AttributeStorage attributeStorage, ObjectStorage objectStorage)
    {
        this.attributeStorage = attributeStorage;
        this.objectStorage = objectStorage;
        this.compressionService = compressionService;
    }

    public void createStorage() throws AttributeStorageException, ObjectStorageException
    {
        // Create the attribute storage domain and object storage bucket
        //
        attributeStorage.createDatabase(AttributeDatabase.shadow_files.toString());
        objectStorage.createBucket(ObjectBucket.shadow_files.toString());

        logDebug(LOG, "created shadow file storage");
    }

    public void deleteStorage() throws ObjectStorageException, AttributeStorageException
    {
        // Delete the attribute storage domain and object storage bucket
        //
        attributeStorage.deleteDatabase(AttributeDatabase.shadow_files.toString());
        objectStorage.deleteBucket(ObjectBucket.shadow_files.toString());

        logDebug(LOG, "deleted shadow file storage");
    }

    public ShadowFile get(String shadowKey) throws ObjectStorageException, IOException
    {
        // Get the full shadow file from the object storage (can skip the attribute storage)
        //
        ShadowFile result = ShadowFile.fromBytes(
                compressionService.decompress(
                        objectStorage.get(ObjectBucket.shadow_files.toString(), shadowKey)
                )
        );

        logDebug(LOG, "got shadow file %s of size %d", result.getKey(), result.getSize());

        return result;
    }

    public ShadowFile getStub(String shadowKey) throws AttributeStorageException
    {
        // Get the attributes (which may be spread over several bundles) and assemble them into a shadow file stub
        // We only need to worry about the size and fn attributes when creating a stub (not snapshot specific stuff)
        //
        Attributes attrs = coalesceBundles(shadowKey, new ArrayList<String>(Arrays.asList("fn"))).first;

        if (attrs != null)
        {
            ShadowFile result = new ShadowFile(shadowKey, Long.valueOf(attrs.getSingle("size")));
            result.setOriginalFilenames(attrs.get("fn"));   // TODO: consolidate ShadowFile constructors.  clean up ShadowFile in general.

            logDebug(LOG, "got stub for shadow file %s of size %d bytes", result.getKey(), result.getSize());

            return result;
        }
        else
        {
            return null;
        }
    }

    private Attributes putShadowFileToObjectStorageAndGetAttributes(ShadowFile shadowFile, Collection<String> snapshots) throws ObjectStorageException, IOException
    {
        Attributes attrs = new Attributes();

        // Put the shadow file into the object storage if it's new (the originalFileNames is empty), and add the fn attribute
        //
        if (shadowFile.getOriginalFilenames() == null)
        {
            objectStorage.put(ObjectBucket.shadow_files.toString(), shadowFile.getKey(), compressionService.compress(shadowFile.getBytes()));
            shadowFile.setOriginalFilenames(shadowFile.getFilenames());
            attrs.add("fn", shadowFile.getOriginalFilenames());
        }

        for (String snapshot : snapshots)
        {
            // Encode the date into the snapshot attribute name.  If there's already an attribute for this snapshot from
            // a previous failed run, then we'll be adding a second with a newer date here, and effectively orphan the previous
            //
            attrs.add("snapshot", snapshot + SNAPSHOT_DATE_DELIM + DateFormatUtils.format(shadowFile.getDate(), "yyyyMMddHHmmss"));

            // If the filenames for this snapshot doesn't match the shadow file's original filenames,
            // then add attributes specifying a set of override filenames
            //
            if (!shadowFile.getOriginalFilenames().equals(shadowFile.getFilenames()))
            {
                attrs.add(snapshot + "-fn", shadowFile.getFilenames());
            }
        }

        return attrs;
    }

    public void batchPut(Collection<ShadowFile> shadowFiles, Collection<String> snapshots) throws ObjectStorageException, AttributeStorageException, IOException
    {
        Map<ShadowFile, Attributes> attributes = new HashMap<ShadowFile, Attributes>();

        for (ShadowFile shadowFile : shadowFiles)
        {
            // Put the shadow file to the object storage and add attribute bundles to this batch
            //
            attributes.put(shadowFile, putShadowFileToObjectStorageAndGetAttributes(shadowFile, snapshots));
        }

        // Put the bundles with a batch put operation on the attribute storage
        //
        batchPutBundles(attributes);
    }

    public void put(ShadowFile shadowFile, Collection<String> snapshots) throws ObjectStorageException, IOException, AttributeStorageException
    {
        // Put the shadow file to the object storage and put attribute bundles to the attribute storage
        //
        putBundles(shadowFile, putShadowFileToObjectStorageAndGetAttributes(shadowFile, snapshots));

        logDebug(LOG, "put shadow file %s in snapshot %s", shadowFile.getKey(), StringUtils.join(snapshots, ", "));
    }

    public ShadowFile delete(String shadowKey, Collection<String> snapshots) throws ObjectStorageException, AttributeStorageException, IOException
    {
        // Get all the attributes for this shadow file from the attribute storage (may be spread across several bundles)
        // We only need to get the snapshot and snapshot-specific attributes
        //
        List<String> attributesToGet =  new ArrayList<String>();
        attributesToGet.add("snapshot");
        for (String snapshot : snapshots)
        {
            attributesToGet.add(snapshot + "-fn");
        }
        Pair<Attributes, Integer> pair = coalesceBundles(shadowKey, attributesToGet);

        final Attributes attrs = pair.first;
        final int bundleCount = pair.second;

        if (attrs != null)
        {
            // Build up attributes that we want to delete
            //
            Set<String> snapshotSet = new HashSet<String>(snapshots);
            Attributes toDelete = new Attributes();

            for (String v : new ArrayList<String>(attrs.get("snapshot")))
            {
                final String snapshot = StringUtils.substringBeforeLast(v, SNAPSHOT_DATE_DELIM);

                // If we're deleting this shadow file from this snapshot, add the corresponding attributes to the list
                // of attributes to delete, including any snapshot specific attributes
                //
                if (snapshotSet.isEmpty() || snapshotSet.contains(snapshot))
                {
                    toDelete.add("snapshot", attrs.remove("snapshot", v));

                    for (Map.Entry<String, Set<String>> entry : attrs.removeWithPrefix(snapshot).entrySet())
                    {
                        toDelete.add(entry.getKey(), entry.getValue());
                    }
                }
            }

            // If other snapshots are still referencing this shadow file, only delete the relevant attributes compiled above,
            // otherwise delete everything (including the shadow file object in the object storage)
            //
            if (attrs.get("snapshot") != null)
            {
                // Delete from all the attribute bundles
                //
                for (int i = 0; i < bundleCount; i++)
                {
                    attributeStorage.delete(AttributeDatabase.shadow_files.toString(), shadowKey + "-" + i, toDelete);
                }

                logDebug(LOG, "removed shadow file %s from snapshot %s", shadowKey, StringUtils.join(snapshots, ", "));

                // Return null to indicate that the shadow file was not deleted
                //
                return null;
            }
            else
            {
                // Get the whole shadow file because we need the list of chunks, so we can dereference this shadow file
                // from those chunks.  The client code will take care of this via ChunkService
                //
                ShadowFile shadowFile = get(shadowKey);

                // Delete all the attribute bundles
                //
                for (int i = 0; i < bundleCount; i++)
                {
                    attributeStorage.delete(AttributeDatabase.shadow_files.toString(), shadowKey + "-" + i);
                }

                // Delete the shadow file object
                //
                objectStorage.delete(ObjectBucket.shadow_files.toString(), shadowKey);

                logDebug(LOG, "deleted shadow file %s", shadowKey);

                // Return the shadow file to indicate that it was deleted
                //
                return shadowFile;
            }
        }
        else
        {
            // Return null to indicate that the shadow file was not deleted
            //
            return null;
        }
    }

    public String createShadowKey(InputStream in) throws NoSuchAlgorithmException, IOException
    {
        // We'll read the file a megabyte at a time and calculate a SHA-1 hash of the whole contents, then append the
        // size to form the shadow key
        //
        String result;
        MessageDigest sha = MessageDigest.getInstance("SHA-1");

        byte[] buffer = new byte[1000000];
        int totalBytes = 0;

        try
        {
            int bytesRead;
            while ((bytesRead = in.read(buffer)) >= 0)
            {
                sha.update(buffer, 0, bytesRead);
                totalBytes += bytesRead;
            }

            result = new String(Hex.encodeHex(sha.digest())) + "-" + totalBytes;
        }
        finally
        {
            in.close();
        }

        return result;
    }

    // TODO: need to handle case where snapshot attribute and matching fn attribute are in different bundles
    public void findStubs(final String snapshot, final String filenamePrefix, final ResultConsumer<ShadowFile> consumer) throws AttributeStorageException, ParseException
    {
        AttributeStorageQuery.Condition condition = new AttributeStorageQuery.LeafCondition("snapshot", AttributeStorageQuery.Operator.like, snapshot + SNAPSHOT_DATE_DELIM + "%");

        if (StringUtils.isNotEmpty(filenamePrefix))
        {
            // Because shadow files can have their attributes spread across multiple bundles, there's a problem here:
            // we might get false negatives with this intersection operation -- it is AND-deficient!
            //
            condition = new AttributeStorageQuery.ComplexCondition(condition,
                                                                   AttributeStorageQuery.Conjunction.and,
                                                                   new AttributeStorageQuery.ComplexCondition(new AttributeStorageQuery.LeafCondition(snapshot + "-fn", AttributeStorageQuery.Operator.like, filenamePrefix + "%"),
                                                                                                              AttributeStorageQuery.Conjunction.or,
                                                                                                              new AttributeStorageQuery.ComplexCondition(new AttributeStorageQuery.NullCondition(snapshot + "-fn"),
                                                                                                                                                         AttributeStorageQuery.Conjunction.and,
                                                                                                                                                         new AttributeStorageQuery.LeafCondition("fn",
                                                                                                                                                                                                 AttributeStorageQuery.Operator.like,
                                                                                                                                                                                                 filenamePrefix + "%"))));
        }

        // TODO: order by
//        query += " order by snapshot desc";

        attributeStorage.find(AttributeDatabase.shadow_files.toString(), new AttributeStorageQuery(condition), new ResultConsumer<Pair<String, Attributes>>()
        {
            public void newResult(Pair<String, Attributes> result) throws Exception
            {
                // The resulting attributes might be only one bundle of many for this shadow file.  So as a heurestic:
                //
                //  1. If it is not the first bundle, obviously there are other bundles so we need to coalesce
                //  2. If there are > 200 attributes then there's a good chance that there are other bundles we need to coalesce
                //
                // TODO: we need to fix this because this heuristic is not fool-proof (particularly if there were attributes deleted leaving a bundle with < 200)
                //  - maybe in delete we can detect if this happens and add an attr indicating that a batch needs coalescing
                String key = StringUtils.substringBeforeLast(result.first, "-");
                if (!StringUtils.endsWith(result.first, "-0") || result.second.size() >= 200)
                {
                    logDebug(LOG, "coalescing possible multi-bundle item: %s", key);

                    result.second = coalesceBundles(key, null).first;
                }

                // Convert the attributes to a shadow file stub, and if there were any matching filenames then add to result
                // If toStub returns null then that means this snapshot name didn't actually match
                //
                ShadowFile stub = toStub(snapshot, filenamePrefix, key, result.second);
                if (consumer != null && stub != null && !stub.getFilenames().isEmpty())
                {
                    consumer.newResult(stub);
                }
            }
        });
    }

    private ShadowFile toStub(String snapshot, String filenamePrefix, String shadowKey, Attributes attrs) throws ParseException
    {
        // Grab the date and time that is encoded in the snapshot name.  If there are multiple, use the most recent
        //
        Date date = null;
        List<String> snapshots = new ArrayList<String>(attrs.get("snapshot"));
        Collections.sort(snapshots, Collections.reverseOrder());
        SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmmss");
        for (String v : snapshots)
        {
            if (v.startsWith(snapshot + SNAPSHOT_DATE_DELIM))
            {
                date = sdf.parse(StringUtils.substringAfterLast(v, SNAPSHOT_DATE_DELIM));
                break;
            }
        }

        // If we didn't find the date then that means this snapshot wasn't in attrs, so return null
        //
        if (date == null)
        {
            return null;
        }

        // Grab the size from the size attribute
        //
        final ShadowFile result = new ShadowFile(shadowKey, Long.valueOf(attrs.getSingle("size")), date);

        // Look for snapshot specific filenames (snapshot-fn), fall back on original filenames (fn)
        //
        Set<String> filenames = attrs.get(snapshot + "-fn");
        if (filenames == null || filenames.isEmpty())
        {
            filenames = attrs.get("fn");
        }

        // Filter filenames by prefix
        //
        for (String filename : filenames)
        {
            if (StringUtils.isEmpty(filenamePrefix) || StringUtils.startsWith(filename, filenamePrefix))
            {
                result.addFile(filename);
            }
        }

        return result;
    }

    private void batchPutBundles(Map<ShadowFile, Attributes> attributes) throws AttributeStorageException
    {
        // Do batch puts on the attribute storage which will remove those attributes that were successfully put
        // For ones that fail, put them into the next bundle, and so on until they're all put
        //
        for (int i = 0; !attributes.isEmpty(); i++)
        {
            Map<String, Attributes> batch = new HashMap<String, Attributes>();

            for (Map.Entry<ShadowFile, Attributes> entry : attributes.entrySet())
            {
                ShadowFile shadowFile = entry.getKey();
                Attributes attrs = entry.getValue();

                // Put the size attribute in every bundle because we need some attribute that indicates a bundle exists
                // when coalescing
                //
                attrs.add("size", Long.toString(shadowFile.getSize()));
                attrs.setPriority("size", 1);
                batch.put(shadowFile.getKey() + "-" + i, attrs);
            }

            attributeStorage.batchPut(AttributeDatabase.shadow_files.toString(), batch);

            // Remove those that were successfully put
            //
            for (ShadowFile shadowFile : new ArrayList<ShadowFile>(attributes.keySet()))
            {
                if (!batch.containsKey(shadowFile.getKey() + "-" + i))
                {
                    attributes.remove(shadowFile);
                }
            }
        }
    }

    private void putBundles(ShadowFile shadowFile, Attributes attributes) throws AttributeStorageException
    {
        // Keep putting to attribute storage in sequential bundles until all attributes have been put
        //
        for (int i = 0; !attributes.isEmpty(); i++)
        {
            // Put the size attribute in every bundle because we need some attribute that indicates a bundle exists
            // when coalescing
            //
            attributes.add("size", Long.toString(shadowFile.getSize()));
            attributes.setPriority("size", 1);
            attributeStorage.put(AttributeDatabase.shadow_files.toString(), shadowFile.getKey() + "-" + i, attributes);
        }
    }

    private Pair<Attributes, Integer> coalesceBundles(String key, List<String> attributesToGet) throws AttributeStorageException
    {
        // Get an array of attributes to get, but not too many at once (< 128)
        //
        String[] attributesToGetArr = null;
        if (attributesToGet != null && !attributesToGet.isEmpty() && attributesToGet.size() < 128)
        {
            attributesToGet.add("size");
            attributesToGetArr = new String[attributesToGet.size()];
            for (int i = 0; i < attributesToGet.size(); i++)
            {
                attributesToGetArr[i] = attributesToGet.get(i);
            }
        }

        Attributes result = new Attributes();

        // Sequentially get bundles until there are no more, and keep track of how many there were
        // We know every bundle result will be non-empty if it exists, because every bundle has the size attribute
        //
        int i = 0;
        Attributes bundle;
        do
        {
            bundle = attributeStorage.get(AttributeDatabase.shadow_files.toString(), key + "-" + i++, attributesToGetArr);
            result.merge(bundle);
        }
        while (bundle != null);

        return Pair.get(result.isEmpty() ? null : result, i);
    }
}
